package cqupt.spark_log.info;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import cqupt.spark_log.contrant.Contant;
import cqupt.spark_log.entity.Executor;
import cqupt.spark_log.entity.Job;
import cqupt.spark_log.entity.Stage;
import cqupt.spark_log.entity.Task;
import cqupt.spark_log.fileio.FileReadMan;
import cqupt.spark_log.utils.Utils;

public class BanchInfo {
	private Map<String, Executor> executors = new HashMap<String, Executor>();
	private Map<String, Job> jobs = new HashMap<String, Job>();
	private Map<String, Stage> stages = new HashMap<String, Stage>();
	private Map<String, Task> tasks = new HashMap<String, Task>();
	private FileReadMan fileTools;
	private String curentJob = null;
	private long makeSpanStart = 0;
	private long makeSpanEnd = 0;
	
	public BanchInfo() {
		// 读取bench.log,分析当前运算的阶段关系及时间
		this.fileTools = new FileReadMan("/bench.log",true);
	}

	public void exe() {
		while (fileTools.hasNext()) {
			// 获得任务启动后的计算资源
			String content = fileTools.next();
			if (content.contains("StandaloneSchedulerBackend: Granted")) {
				int executorIdIndex = content.indexOf("executor ID");
				int onhostPortIndex = content.indexOf("on hostPort");
				int withIndex = content.indexOf("with");
				StringBuilder stringBuilder = new StringBuilder(content);
				String executorId = stringBuilder.substring(executorIdIndex + 11, onhostPortIndex).trim();
				String hostPort = stringBuilder.substring(onhostPortIndex + 11, withIndex).trim().split(":")[0];
				String source = stringBuilder.substring(withIndex + 4, stringBuilder.length()).trim();
				String cores = "";
				String ram = "";
				String[] sources = source.split(",");
				for (int i = 0; i < sources.length; i++) {
					String temp = sources[i];
					if (temp.contains("cores")) {
						cores = temp.replaceAll("cores", "").trim();
					} else if (temp.contains("RAM")) {
						ram = temp.replaceAll("RAM", "").trim();
					}
				}
				Executor executor = new Executor(executorId, hostPort, cores, ram);
				String ip = hostPort.split(":")[0].trim();
				executors.put(ip, executor);
			}
			// 获得job开始
			if (content.contains("DAGScheduler: Got job")) {
				Date startTime = Utils.getTime(content);
				int jobIndex = content.indexOf("job");
				int jobendIndex = content.indexOf("(");
				String jobId = content.substring(jobIndex + 3, jobendIndex).trim();
				Job job = new Job(jobId, startTime);
				jobs.put(jobId, job);
				curentJob = jobId;
			}
			// 获得job结束
			if (content.matches(".*DAGScheduler: Job.*finished.*")) {
				Date endTime = Utils.getTime(content);
				int jobIndex = content.indexOf("Job");
				int jobendIndex = content.indexOf("finished");
				String jobId = content.substring(jobIndex + 3, jobendIndex).trim();
				Job job = jobs.get(jobId);
				job.setEndTime(endTime);
				curentJob = null;
			}
			// 获取阶段开始
			if (content.matches(".*DAGScheduler: Submitting(?!.*tasks).*")) {
				int beginIndex = 0;
				String stageType = "";
				if (content.contains("ShuffleMapStage")) {
					stageType = "ShuffleMapStage";
					beginIndex = content.indexOf("ShuffleMapStage") + 15;
				} else if (content.contains("ResultStage")) {
					stageType = "ResultStage";
					beginIndex = content.indexOf("ResultStage") + 11;
				}
				int endIndex = content.indexOf("(");
				String stageId = content.substring(beginIndex, endIndex).trim();
				Date startTime = Utils.getTime(content);
				Stage stage = new Stage(stageId, stageType, startTime, curentJob);
				stages.put(stageId, stage);
				Job job = jobs.get(curentJob);
				job.addStage(stage);
			}
			// 获取阶段结束
			if (content.matches(".*DAGScheduler:.*Stage.*finished.*")) {
				int beginIndex = content.indexOf("Stage") + 5;
				int endIndex = content.indexOf("(");
				String stageId = content.substring(beginIndex, endIndex).trim();
				Date endTime = Utils.getTime(content);
				Stage stage = stages.get(stageId);
				stage.setEndTime(endTime);
			}
			// 任务开始
			if (content.matches(".*TaskSetManager: Starting task.*")) {
				int beginStageIndex = content.indexOf("stage") + 5;
				int beginTaskIndex = content.indexOf("task") + 4;
				int endTaskIndex = content.indexOf("in stage");
				int sIndex = content.indexOf("(");
				int eIndex = content.indexOf(")");
				String[] context = content.substring(sIndex + 1, eIndex).trim().split(",");
				String taskId = context[0].replace("TID", "").trim();
				String executorIp = context[1].trim();
				String stageId = content.substring(beginStageIndex, sIndex).trim();
				String[] stageIds = stageId.split("\\.");
				stageId = stageIds[0];
				String index = content.substring(beginTaskIndex, endTaskIndex).trim();
				String[] indexs = index.split("\\.");
				index = indexs[0];
				Date startTime = Utils.getTime(content);
				Task task = new Task(taskId, index, startTime, Contant.APP_NAME, curentJob, stageId, executorIp);
				tasks.put(taskId, task);
				Stage stage = stages.get(stageId);
				stage.addTask(task);
			}
			// 任务结束
			if (content.matches(".*TaskSetManager: Finished task.*")) {
				Pattern pattern = Pattern.compile("\\d* ms");
				Matcher matcher = pattern.matcher(content);
				String runTime = "0";
				if (matcher.find()) {
					runTime = matcher.group(0).replace("ms", "").trim();
				}
				int sIndex = content.indexOf("(");
				int eIndex = content.indexOf(")");
				String context = content.substring(sIndex + 1, eIndex).trim();
				String taskId = context.replace("TID", "").trim();
				Date endTime = Utils.getTime(content);
				Task task = tasks.get(taskId);
				task.setEndTime(endTime);
				task.setRunTime(Integer.parseInt(runTime));
			}
			// 获取任务附加时间
			if (content.matches(".*TaskSetManager: tasiId.*")) {
				int index = content.indexOf("TaskSetManager:")+15;
				String line = content.substring(index);
				String[] valuelines = line.split(";");
				String taskId = valuelines[0].split(":")[1].trim();
				Task task = tasks.get(taskId);
				String executorDeserializeTime = valuelines[1].split(":")[1].trim();
				task.setExecutorDeserializeTime(Long.parseLong(executorDeserializeTime));
				String executorDeserializeCpuTime = valuelines[2].split(":")[1].trim();
				task.setExecutorDeserializeCpuTime(Long.parseLong(executorDeserializeCpuTime));
				String executorRunTime = valuelines[3].split(":")[1].trim();
				task.setExecutorRunTime(Long.parseLong(executorRunTime));
				String executorCpuTime = valuelines[4].split(":")[1].trim();
				task.setExecutorCpuTime(Long.parseLong(executorCpuTime));
				String jvmGCTime = valuelines[5].split(":")[1].trim();
				task.setJvmGCTime(Long.parseLong(jvmGCTime));
				String resultSerializationTime = valuelines[6].split(":")[1].trim();
				task.setResultSerializationTime(Long.parseLong(resultSerializationTime));
			}
			
			//app开始时间
			if (content.matches(".*SparkUI: Bound SparkUI.*")) {
				Date appStartTime = Utils.getTime(content);
				makeSpanStart = appStartTime.getTime();
			}
			
			//app结束
			if (content.matches(".*SparkUI: Stopped.*")) {
				Date appStartTime = Utils.getTime(content);
				makeSpanEnd = appStartTime.getTime();
			}
		}
	}

	public Map<String, Executor> getExecutors() {
		return executors;
	}

	public Map<String, Job> getJobs() {
		return jobs;
	}

	public Map<String, Stage> getStages() {
		return stages;
	}

	public Map<String, Task> getTasks() {
		return tasks;
	}
	
	public long getMakeSpan() {
		long result = 0;
		//job是串行的
		for (String key : jobs.keySet()) {
			Job job = jobs.get(key);
			result += job.getRuntime();
		}
		return result;
	}

	public long getALLMakeSpan() {
		return getMakeSpanEnd()-getMakeSpanStart();
	}

	public long getMakeSpanStart() {
		return makeSpanStart;
	}

	public void setMakeSpanStart(long makeSpanStart) {
		this.makeSpanStart = makeSpanStart;
	}

	public long getMakeSpanEnd() {
		return makeSpanEnd;
	}

	public void setMakeSpanEnd(long makeSpanEnd) {
		this.makeSpanEnd = makeSpanEnd;
	}
	
	
}
